package com.mqtt.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.mqtt.model.DeviceInfo;
import com.mqtt.model.DeviceSession;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 设备会话管理服务
 */
@Slf4j
@Service
public class DeviceSessionService {
    
    @Autowired
    private JedisPool jedisPool;
    
    // 本地缓存
    private final Map<String, DeviceSession> sessionCache = new ConcurrentHashMap<>();
    
    private static final String SESSION_KEY_PREFIX = "device:session:";
    private static final String ONLINE_DEVICES_KEY = "device:online";
    private static final int SESSION_EXPIRE_TIME = 3600; // 1小时
    private static final int HEARTBEAT_TIMEOUT = 180; // 3分钟
    
    /**
     * 创建设备会话
     */
    public void createSession(String deviceId, JSONObject deviceInfo) {
        try {
            DeviceSession session = new DeviceSession();
            session.setDeviceId(deviceId);
            session.setDeviceType(deviceInfo.getString("deviceType"));
            session.setVersion(deviceInfo.getString("version"));
            session.setIpAddress(deviceInfo.getString("ipAddress"));
            session.setConnectTime(new Date());
            session.setLastActiveTime(new Date());
            session.setStatus("online");
            session.setAttributes(deviceInfo.getInnerMap());
            
            // 保存到Redis
            saveSessionToRedis(session);
            
            // 保存到本地缓存
            sessionCache.put(deviceId, session);
            
            // 添加到在线设备集合
            addToOnlineDevices(deviceId);
            
            log.info("Device session created: {}", deviceId);

        } catch (Exception e) {
            log.error("Failed to create session for device: {}", deviceId, e);
        }
    }
    
    /**
     * 更新设备会话
     */
    public void updateSession(String deviceId, String topic) {
        try {
            DeviceSession session = getSession(deviceId);
            if (session == null) {
                // 如果会话不存在，创建新会话
                JSONObject defaultInfo = new JSONObject();
                defaultInfo.put("deviceType", "unknown");
                defaultInfo.put("version", "unknown");
                createSession(deviceId, defaultInfo);
                session = getSession(deviceId);
            }
            
            if (session != null) {
                session.setLastActiveTime(new Date());
                session.setLastTopic(topic);
                
                // 更新到Redis
                saveSessionToRedis(session);
                
                // 更新本地缓存
                sessionCache.put(deviceId, session);
            }
            
        } catch (Exception e) {
            log.error("Failed to update session for device: {}", deviceId, e);
        }
    }
    
    /**
     * 更新设备状态
     */
    public void updateDeviceStatus(String deviceId, String status) {
        try {
            DeviceSession session = getSession(deviceId);
            if (session != null) {
                session.setStatus(status);
                session.setLastActiveTime(new Date());
                
                saveSessionToRedis(session);
                sessionCache.put(deviceId, session);
                
                if ("offline".equals(status)) {
                    removeFromOnlineDevices(deviceId);
                } else if ("online".equals(status)) {
                    addToOnlineDevices(deviceId);
                }
                
                log.debug("Device {} status updated to: {}", deviceId, status);
            }
            
        } catch (Exception e) {
            log.error("Failed to update status for device: {}", deviceId, e);
        }
    }
    
    /**
     * 更新心跳时间
     */
    public void updateHeartbeat(String deviceId) {
        try {
            DeviceSession session = getSession(deviceId);
            if (session != null) {
                session.setLastHeartbeatTime(new Date());
                session.setLastActiveTime(new Date());
                
                saveSessionToRedis(session);
                sessionCache.put(deviceId, session);
                
                log.debug("Heartbeat updated for device: {}", deviceId);
            }
            
        } catch (Exception e) {
            log.error("Failed to update heartbeat for device: {}", deviceId, e);
        }
    }
    
    /**
     * 移除设备会话
     */
    public void removeSession(String deviceId) {
        try {
            // 从Redis删除
            try (Jedis jedis = jedisPool.getResource()) {
                jedis.del(SESSION_KEY_PREFIX + deviceId);
            }
            
            // 从本地缓存删除
            sessionCache.remove(deviceId);
            
            // 从在线设备集合移除
            removeFromOnlineDevices(deviceId);
            
            log.info("Device session removed: {}", deviceId);
            
        } catch (Exception e) {
            log.error("Failed to remove session for device: {}", deviceId, e);
        }
    }
    
    /**
     * 获取设备会话
     */
    public DeviceSession getSession(String deviceId) {
        // 先从本地缓存获取
        DeviceSession session = sessionCache.get(deviceId);
        if (session != null) {
            return session;
        }
        
        // 从Redis获取
        try (Jedis jedis = jedisPool.getResource()) {
            String sessionJson = jedis.get(SESSION_KEY_PREFIX + deviceId);
            if (sessionJson != null) {
                session = JSON.parseObject(sessionJson, DeviceSession.class);
                // 放入本地缓存
                sessionCache.put(deviceId, session);
                return session;
            }
        } catch (Exception e) {
            log.error("Failed to get session for device: {}", deviceId, e);
        }
        
        return null;
    }
    
    /**
     * 获取在线设备列表
     */
    public List<DeviceInfo> getOnlineDevices() {
        List<DeviceInfo> devices = new ArrayList<>();
        
        try (Jedis jedis = jedisPool.getResource()) {
            Set<String> deviceIds = jedis.smembers(ONLINE_DEVICES_KEY);
            
            for (String deviceId : deviceIds) {
                DeviceInfo deviceInfo = getDeviceInfo(deviceId);
                if (deviceInfo != null) {
                    devices.add(deviceInfo);
                }
            }
            
        } catch (Exception e) {
            log.error("Failed to get online devices", e);
        }
        
        return devices;
    }
    
    /**
     * 检查设备是否在线
     */
    public boolean isDeviceOnline(String deviceId) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.sismember(ONLINE_DEVICES_KEY, deviceId);
        } catch (Exception e) {
            log.error("Failed to check device online status: {}", deviceId, e);
            return false;
        }
    }
    
    /**
     * 获取设备信息
     */
    public DeviceInfo getDeviceInfo(String deviceId) {
        DeviceSession session = getSession(deviceId);
        if (session == null) {
            return null;
        }
        
        DeviceInfo deviceInfo = new DeviceInfo();
        deviceInfo.setDeviceId(session.getDeviceId());
        deviceInfo.setDeviceType(session.getDeviceType());
        deviceInfo.setVersion(session.getVersion());
        deviceInfo.setStatus(session.getStatus());
        deviceInfo.setIpAddress(session.getIpAddress());
        deviceInfo.setConnectTime(session.getConnectTime());
        deviceInfo.setLastActiveTime(session.getLastActiveTime());
        
        return deviceInfo;
    }
    
    /**
     * 保存会话到Redis
     */
    private void saveSessionToRedis(DeviceSession session) {
        try (Jedis jedis = jedisPool.getResource()) {
            String key = SESSION_KEY_PREFIX + session.getDeviceId();
            String value = JSON.toJSONString(session);
            jedis.setex(key, SESSION_EXPIRE_TIME, value);
        } catch (Exception e) {
            log.error("Failed to save session to Redis: {}", session.getDeviceId(), e);
        }
    }
    
    /**
     * 添加到在线设备集合
     */
    private void addToOnlineDevices(String deviceId) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.sadd(ONLINE_DEVICES_KEY, deviceId);
        } catch (Exception e) {
            log.error("Failed to add device to online set: {}", deviceId, e);
        }
    }
    
    /**
     * 从在线设备集合移除
     */
    private void removeFromOnlineDevices(String deviceId) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.srem(ONLINE_DEVICES_KEY, deviceId);
        } catch (Exception e) {
            log.error("Failed to remove device from online set: {}", deviceId, e);
        }
    }
    
    /**
     * 定时检查设备心跳超时
     */
    @Scheduled(fixedDelay = 60000) // 每分钟执行一次
    public void checkHeartbeatTimeout() {
        log.debug("Checking device heartbeat timeout...");
        
        List<String> timeoutDevices = new ArrayList<>();
        long currentTime = System.currentTimeMillis();
        
        for (DeviceSession session : sessionCache.values()) {
            if (session.getLastHeartbeatTime() != null) {
                long lastHeartbeat = session.getLastHeartbeatTime().getTime();
                if (currentTime - lastHeartbeat > HEARTBEAT_TIMEOUT * 1000) {
                    timeoutDevices.add(session.getDeviceId());
                }
            }
        }
        
        // 处理超时设备
        for (String deviceId : timeoutDevices) {
            log.warn("Device {} heartbeat timeout", deviceId);
            updateDeviceStatus(deviceId, "offline");
        }
    }
    
    /**
     * 获取设备统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        
        try {
            List<DeviceInfo> onlineDevices = getOnlineDevices();
            
            stats.put("totalOnline", onlineDevices.size());
            stats.put("totalCached", sessionCache.size());
            
            // 按设备类型统计
            Map<String, Long> typeCount = onlineDevices.stream()
                    .collect(Collectors.groupingBy(
                            DeviceInfo::getDeviceType,
                            Collectors.counting()
                    ));
            stats.put("deviceTypes", typeCount);
            
        } catch (Exception e) {
            log.error("Failed to get statistics", e);
        }
        
        return stats;
    }
}